package com.usian.behavior.behavior.listner;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.usian.behavior.behavior.controller.BehaviorEntryContorller;
import com.usian.behavior.behavior.mapper.BehaviorEntryMapperMapper;
import com.usian.behavior.behavior.mapper.FollowBehaviorMapper;
import com.usian.common.medis.NewsAutoScanConstants;
import com.usian.model.behavior.pojo.ApBehaviorEntry;
import com.usian.model.behavior.pojo.ApFollowBehavior;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Map;
import java.util.Optional;

@Component
public class BehaviorListner {

    @Autowired
    private BehaviorEntryContorller behaviorEntryContorller;

    @Autowired
    private FollowBehaviorMapper followBehaviorMapper;

    @Autowired
    private BehaviorEntryMapperMapper behaviorEntryMapperMapper;


    //kakfa监听的主题
    @KafkaListener(topics = NewsAutoScanConstants.AP_BEHAVIOR)
    public void dddddd(ConsumerRecord<?,?> record){
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if(optional.isPresent()){
            Object value = record.value();

            String s = value.toString();
            Map<String, String> map = JSONObject.parseObject(s, new TypeReference<Map<String, String>>() {});

            LambdaQueryWrapper<ApBehaviorEntry> q = new LambdaQueryWrapper<>();
            q.eq(ApBehaviorEntry::getEntryId,Integer.parseInt(map.get("userId")));
            ApBehaviorEntry apBehaviorEntry1 = behaviorEntryMapperMapper.selectOne(q);
            if(apBehaviorEntry1==null){
                apBehaviorEntry1.setCreatedTime(new Date());
                apBehaviorEntry1.setEntryId(Integer.parseInt(map.get("userId")));
                apBehaviorEntry1.setType(1);
                behaviorEntryContorller.add(apBehaviorEntry1);
            }

            LambdaQueryWrapper<ApFollowBehavior> qq = new LambdaQueryWrapper<>();
            qq.eq(ApFollowBehavior::getEntryId,apBehaviorEntry1.getId());
            qq.eq(ApFollowBehavior::getFollowId,map.get("followId"));
            qq.eq(ApFollowBehavior::getArticleId,map.get("articleId"));
            ApFollowBehavior apFollowBehavior1 = followBehaviorMapper.selectOne(qq);
            ApFollowBehavior apFollowBehavior = new ApFollowBehavior();
            if(apFollowBehavior1==null){
                apFollowBehavior.setArticleId(Long.parseLong(map.get("articleId")));
                apFollowBehavior.setCreatedTime(new Date());
                apFollowBehavior.setEntryId(apBehaviorEntry1.getId());
                apFollowBehavior.setFollowId(Integer.parseInt(map.get("followId")));
                followBehaviorMapper.insert(apFollowBehavior);
            }
        }
    }
}
